package com.umeox.watch.moto.dataservice.mqtt.impl;

import android.content.Context;
import android.text.TextUtils;

import com.alibaba.fastjson.JSONObject;
import com.umeox.moto.common.log.Logger;
import com.umeox.moto.mqtt.MqttConfigure;
import com.umeox.moto.mqtt.MqttInitParams;
import com.umeox.moto.mqtt.PersistentNet;
import com.umeox.moto.mqtt.base.AError;
import com.umeox.moto.mqtt.base.ARequest;
import com.umeox.moto.mqtt.base.AResponse;
import com.umeox.moto.mqtt.base.IOnCallListener;
import com.umeox.moto.mqtt.base.PersistentConnectState;
import com.umeox.moto.mqtt.event.IConnectionStateListener;
import com.umeox.moto.mqtt.event.IOnRrpcResponseHandle;
import com.umeox.moto.mqtt.event.IOnSubscribeListener;
import com.umeox.moto.mqtt.event.IOnSubscribeRpcListener;
import com.umeox.moto.mqtt.event.PersistentEventDispatcher;
import com.umeox.moto.mqtt.request.PersistentRequest;
import com.umeox.watch.moto.dataservice.mqtt.ConnectConfig;
import com.umeox.watch.moto.dataservice.mqtt.Topic;
import com.umeox.watch.moto.dataservice.mqtt.interfaces.IMqttClient;
import com.umeox.watch.moto.dataservice.mqtt.interfaces.IRequestListener;
import com.umeox.watch.moto.dataservice.mqtt.request.PublishRequestFactory;

import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.MqttException;

import java.util.concurrent.atomic.AtomicBoolean;

public class MqttClientWrapper implements IMqttClient {
    private final String TAG = "MqttClientWrapper";
    private final AtomicBoolean initialized;
    private Context mContext;
    private OnSubscribeRpcListener subscribeListener;

    public MqttClientWrapper(Context context) {
        Logger.i("init mqttClient");
        mContext = context;
        initialized = new AtomicBoolean(false);
        subscribeListener = new OnSubscribeRpcListener();
    }

    @Override
    public synchronized void startConnect(ConnectConfig configure, IConnectionStateListener listener) {
        Logger.d(TAG, "startConnect()," + configure);
        if (configure != null && configure.checkValid()) {
            if (!initialized.compareAndSet(false, init(configure, listener))) {
                Logger.d(TAG, "startConnect(), channel is connecting or initialized");
            }
        } else {
            Logger.e(TAG, "startConnect(), param error, config is empty");
        }
    }

    private synchronized boolean init(ConnectConfig configure, IConnectionStateListener listener) {
        MqttConfigure.mqttHost = configure.channelHost;
        MqttConfigure.mqttUserName = configure.userName;
        MqttConfigure.mqttClientId = configure.clientId;
        MqttConfigure.mqttPassword = configure.password;
        MqttConfigure.setKeepAliveInterval(configure.keepAliveIntervalSec);
        MqttInitParams initParams = new MqttInitParams(configure.clientId, configure.password);
        initParams.receiveOfflineMsg = true;
        PersistentEventDispatcher.getInstance().registerOnTunnelStateListener(listener, true);
        try {
            PersistentNet.getInstance().init(mContext, initParams);
        } catch (MqttException e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    @Override
    public PersistentConnectState getConnectState() {
        return PersistentNet.getInstance().getConnectState();
    }

    @Override
    public void subscrbie(String topic) {
        if (TextUtils.isEmpty(topic)) {
            Logger.e(TAG, "subscrbie(), topic is Empty");
        } else if (getConnectState() != PersistentConnectState.CONNECTED) {
            Logger.e(TAG, "subscrbie(), is not initialized");
        } else {
            PersistentNet.getInstance().subscribeRpc(topic, subscribeListener);
        }
    }

    @Override
    public void unSubscrbie(String topic) {
        if (TextUtils.isEmpty(topic)) {
            Logger.e(TAG, "unSubscrbie(), topic is Empty");
        } else if (getConnectState() != PersistentConnectState.CONNECTED) {
            Logger.e(TAG, "subscrbie(), is not initialized");
        } else {
            PersistentNet.getInstance().unSubscribeRpc(topic, subscribeListener);
        }
    }

    @Override
    public void ping(IRequestListener listener) {
        Logger.d(TAG, "asyncSendRequest(), topic = PING");
        if (getConnectState() != PersistentConnectState.CONNECTED) {
            if (listener != null) {
                AError error = new AError();
                error.setCode(AError.AKErrorInvokeNetError);
                error.setMsg("Mqtt is not connected");
                listener.onFailure(PublishRequestFactory.createPingPublishRequest(), error);
            }
        } else {
            PersistentNet.getInstance().asyncSend(PublishRequestFactory.createPingPublishRequest(), new OnPublishListener(listener));
        }
    }


    @Override
    public void asyncSendRequest(String topic, String event, Object params, IRequestListener listener) {
        if (getConnectState() != PersistentConnectState.CONNECTED) {
            if (listener != null) {
                AError error = new AError();
                error.setCode(AError.AKErrorInvokeNetError);
                error.setMsg("MQTT长连接已断开");
                listener.onFailure(PublishRequestFactory.createCommonPublishRequest(topic, params), error);
            }
        } else {
            if (topic.startsWith(Topic.CLOUD_RPC_REPLY_TOPIC_PREFIX)) {
                //RPC response
                PersistentNet.getInstance().asyncSend(PublishRequestFactory.createRpcPublishRequest(topic, event, params), new OnPublishListener(listener));
            } else if (topic.startsWith(Topic.EVENT_PUBLISH_TOPIC_PREFIX)) {
                //事件上报
                PersistentNet.getInstance().asyncSend(PublishRequestFactory.createEventPublishRequest(topic, event, params), new OnPublishListener(listener));
            } else if (topic.startsWith(Topic.GET_ATTRIBUTE_TOPIC_PREFIX)) {
                //获取共享属性
                PersistentNet.getInstance().asyncSend(PublishRequestFactory.createAttributePublishRequest(topic, params), new OnPublishListener(listener));
            } else if (topic.equals(Topic.CLIENT_RPC_TOPIC_PREFIX)) {
                //客户端RPC请求
                if (!TextUtils.isEmpty(event)) {
                    PersistentNet.getInstance().asyncSend(PublishRequestFactory.createClientChatRpcPublishRequest(topic, event, params), new OnPublishListener(listener));
                } else {
                    PersistentNet.getInstance().asyncSend(PublishRequestFactory.createClientRpcPublishRequest(topic, params), new OnPublishListener(listener));
                }
            } else if (topic.startsWith(Topic.CLOUD_PUSH_REPLY_TOPIC_PREFIX)) {
                //回复云端消息
                PersistentNet.getInstance().asyncSend(PublishRequestFactory.createAckPushMsgPublishRequest(topic, (JSONObject) params), null);
            } else if (topic.startsWith(Topic.DEVICE_ATTRIBUTE_POST_PREFIX)) {
                //属性版本上报
                PersistentNet.getInstance().asyncSend(PublishRequestFactory.createClientAttrPostRequest(topic, (JSONObject) params), new OnPublishListener(listener));
            }
        }
    }

    @Override
    public void destroy() {
        if (getConnectState() == PersistentConnectState.CONNECTING || getConnectState() == PersistentConnectState.CONNECTED) {
            PersistentNet.getInstance().destroy();
        }
    }

    @Override
    public void disConnectManually(IMqttActionListener callback) {
        if (getConnectState() == PersistentConnectState.CONNECTING || getConnectState() == PersistentConnectState.CONNECTED) {
            PersistentNet.getInstance().disConnectManually(callback);
        }
    }

    private class OnPublishListener implements IOnCallListener {
        private IRequestListener listener;

        public OnPublishListener(IRequestListener listener) {
            this.listener = listener;
        }

        @Override
        public void onSuccess(ARequest request, AResponse response) {
//            if (request instanceof MqttPublishRequest) {
//                MqttPublishRequest mqttPublishRequest = (MqttPublishRequest) request;
//                Logger.e(TAG, "PublishRequest, onSuccess, topic = " + mqttPublishRequest.topic + ",payloadObj:" + mqttPublishRequest.payloadObj);
//            }
            listener.onSuccess(request, response);
        }

        @Override
        public void onFailed(ARequest request, AError error) {
            this.listener.onFailure(request, error);
        }

        @Override
        public boolean needUISafety() {
            return true;
        }
    }

    private class OnSubscribeRpcListener implements IOnSubscribeRpcListener {

        @Override
        public IOnSubscribeListener getSubscribeListener() {

            return new IOnSubscribeListener() {
                @Override
                public void onSuccess(String topic) {
                    Logger.d(TAG, "afterConnect(),onSuccess, topic=" + topic);
                }

                @Override
                public void onFailed(String topic, AError error) {
                    Logger.d(TAG, "afterConnect(),onFailed, error = " + error.getMsg());
                }

                @Override
                public boolean needUISafety() {
                    return false;
                }
            };
        }

        @Override
        public void onReceived(String topic, PersistentRequest request, IOnRrpcResponseHandle responseHandle) {
            Logger.i(TAG, "messageArrived: Topic=" + topic + ", Message=" + request.payloadObj.toString());
        }

        @Override
        public void onResponseSuccess(String topic) {
            Logger.d(TAG, "afterConnect(),onResponseSuccess, topic = " + topic);
        }

        @Override
        public void onResponseFailed(String topic, AError error) {
            Logger.d(TAG, "afterConnect(),onResponseFailed, error = " + error.getMsg());
        }
    }
}
